package com.poetic.network.flow.process;
import com.poetic.network.flow.domain.UrlViewCount;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <pre>
 *  TODO
 * Created by lianghuikun on 2020-09-15.
 * </pre>
 *
 * @author lianghuikun
 */
public class TopNHotUrls extends KeyedProcessFunction<Long, UrlViewCount, String> {

    private int topSize;

    public TopNHotUrls(int topSize) {
        this.topSize = topSize;
    }

    private ListState<UrlViewCount> listState;

    @Override
    public void open(Configuration parameters) throws Exception {

        ListStateDescriptor<UrlViewCount> desc = new ListStateDescriptor<>("listState", TypeInformation.of(new TypeHint<UrlViewCount>() {
        }));
        this.listState = getRuntimeContext()
                .getListState(desc);
    }

    @Override
    public void processElement(UrlViewCount value, Context ctx, Collector<String> out) throws Exception {

        listState.add(value);

        ctx.timerService().registerEventTimeTimer(value.getWindowEnd() + 1);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

        List<UrlViewCount> viewList = new ArrayList<>();
        Iterable<UrlViewCount> iterable = listState.get();
        iterable.forEach(viewList::add);

        listState.clear();

        List<UrlViewCount> countList = Optional.ofNullable(viewList).orElse(Collections.emptyList())
                .stream()
                .sorted(Comparator.comparingLong(UrlViewCount::getCount).reversed())
                .limit(topSize)
                .collect(Collectors.toList());
        StringBuilder show = new StringBuilder();
        show.append("时间：").append(new Timestamp(timestamp - 1)).append("\n");

        for (int i = 0; i < countList.size(); i++) {
            UrlViewCount view = countList.get(i);
            show.append(" NO").append(i+1)
                    .append(" url:").append(view.getUrl())
                    .append(" 流量").append(view.getCount()).append("\n");
        }
        show.append("=================================\n\n");

        out.collect(show.toString());
        TimeUnit.SECONDS.sleep(1);

    }
}
